package raft

import (
	"errors"
	"go.uber.org/zap"
	"math/rand/v2"
	"sort"
	"time"
	"xin/raft/pb"
)

type Raft struct {
	Id                    uint64         // 节点id
	Term                  uint64         // 任期
	VoteFor               uint64         // vote for
	LeaderId              uint64         // leader id
	peers                 peers          // 集群内所有节点;包含自身
	conf                  *Config        // conf
	role                  Role           // 节点状态
	storage               MetaStorage    // 持久化状态
	tickC                 *time.Ticker   // tick
	tick                  func()         // tick func，根据角色变换，领导人心跳超时，跟随者候选人，选举超时
	randomElectionTimeout int            // 随机选举超时
	electionCount         int            // 选举计数
	heartbeatCount        int            // 心跳计数
	trans                 Transport      // 网络通讯
	logRespCh             chan *logResp  // resp chan 发起异步网络请求后的返回结果
	voteRespCh            chan *voteResp // 投票结果 chan; 每次变为 Candidate 时，重新初始化一个通道；且通道长度为集群长度；保证不会相同的节点重复计票。
	grantedVotes          int            // 收到的选票数
	wal                   *wal           // Write-Ahead Logging
	logger                *zap.Logger    // logger
}

func NewRaft(id uint64, storage MetaStorage, trans Transport, wal *wal, conf *Config, logger *zap.Logger) (*Raft, error) {
	err := conf.Validate()
	if err != nil {
		return nil, err
	}

	ps, err := newPeers(conf.Cluster)
	if err != nil {
		return nil, err
	}

	return &Raft{
		Id:      id,
		conf:    conf,
		peers:   ps,
		role:    Follower,
		trans:   trans,
		storage: storage,
		wal:     wal,
		logger:  logger,
	}, nil
}

// Init 初始化
func (r *Raft) Init() error {
	if err := r.storage.Init(); err != nil {
		return err
	}

	if err := r.trans.Init(); err != nil {
		return err
	}

	if err := r.wal.Init(); err != nil {
		return err
	}

	return nil
}

// Shutdown 关闭
func (r *Raft) Shutdown() error {
	if err := r.storage.Shutdown(); err != nil {
		return err
	}
	if err := r.trans.Shutdown(); err != nil {
		return err
	}
	if err := r.wal.Shutdown(); err != nil {
		return err
	}
	return nil
}

func (r *Raft) Start() {
	// Follower
	r.becomeFollower(1, 0, 0)

	r.run()
}

// run 启动 raft 算法
func (r *Raft) run() {
	r.tickC = time.NewTicker(100 * time.Millisecond)

	for {
		select {
		case <-r.tickC.C: // tick 选举超时 心跳超时
			r.tick()
		case rpc := <-r.trans.Consumer():
			r.handleRpc(rpc) // 接收 RPC 请求
		case resp := <-r.voteRespCh:
			r.handleVoteResp(resp) // 收到投票结果
		case resp := <-r.logRespCh:
			r.handleAppendEntriesResp(resp) // 收到追加日志回复结果
		default:

		}
	}
}

// handleRpc 处理 RPC 请求
func (r *Raft) handleRpc(rpc *RPCReq) {
	switch command := rpc.Command.(type) {
	case *pb.RequestVote: // 请求投票
		r.receiveVoteRpc(command, rpc)
	case *pb.AppendEntries: // 日志复制或者心跳
		r.receiveAppendEntriesRpc(command, rpc)
	}
}

type voteResp struct {
	pb.RequestVoteResp
	id uint64 // 投票节点
}

type logResp struct {
	pb.AppendEntriesResp
	id uint64 // 回复节点id
}

func (r *Raft) handleResp(resp any) {
	switch v := resp.(type) {
	case *voteResp: // 收到请求投票结果
		r.handleVoteResp(v)
	case *logResp: // 收到日志回复
		r.handleAppendEntriesResp(v)
	}
}

// handleVoteResp 处理投票结果
func (r *Raft) handleVoteResp(resp *voteResp) {
	r.logger.Debug(">>> raft.handleVoteResp() 收到投票结果", zap.Uint64("Id", resp.id), zap.Uint64("Term", resp.Term), zap.Bool("VoteGranted", resp.VoteGranted))

	if resp.Term > r.Term {
		r.logger.Warn(">>> 投票节点任期超过本地任期", zap.Uint64("Term", r.Term), zap.Uint64("respTerm", resp.Term))
		// 投票节点任期超过本地任期
		r.becomeFollower(resp.Term, 0, 0)
		return
	}

	if r.role != Candidate {
		// 当前节点非候选节点
		r.logger.Warn("收到投票，当前节点非候选者，忽略", zap.Uint64("id", r.Id), zap.String("role", r.role.String()))
		return
	}

	if resp.VoteGranted {
		// 记票
		r.grantedVotes++

		r.logger.Debug("投票计数", zap.Uint64("from", resp.id), zap.String("role", r.role.String()), zap.Int("grantedVotes", r.grantedVotes))
	}

	if r.grantedVotes > r.peers.quorumSize() {
		r.logger.Info(">>> 选举成功", zap.Uint64("term", r.Term))
		r.becomeLeader()
	}

}

// receiveVoteRpc 收到投票请求
func (r *Raft) receiveVoteRpc(command *pb.RequestVote, rpc *RPCReq) {
	r.logger.Debug(">>> 收到其他节点的投票请求准备开始投票", zap.Uint64("CandidateId", command.CandidateId), zap.Uint64("rpcTerm", command.Term))

	r.randomElectionTime() // 收到请求投票后，重置定时计数。以防当前节点成为候选人

	// 收到其他节点的请求投票
	resp := &pb.RequestVoteResp{
		Term:        command.Term,
		VoteGranted: false,
	}
	var err error

	defer func() {
		r.logger.Debug(">>> 投票结果写入 chan", zap.Uint64("id", r.Id), zap.Bool("VoteGranted", resp.VoteGranted))
		rpc.RespChan <- RPCResp{
			Err:  err,
			Resp: resp,
		}
	}()

	// 请求投票节点任期比当前节点小

	// 请求投票节点宕机好久了刚恢复，然后就进入了选举状态。其他的节点可能已经经过了好几轮选举。任期肯定比他大。
	if command.Term < r.Term {
		r.logger.Debug(">>> 拒绝投票 请求投票节点的任期小于本节点.", zap.Uint64("rpcTerm", command.Term), zap.Uint64("currentTerm", r.Term))

		resp.Term = r.Term
		resp.VoteGranted = false

		return
	}

	// 任期相同
	// 大部分情况是，当前节点已经成为了候选人，给其他节点发了请求投票请求
	// 然后其他的节点也成为了候选人。然后发了请求投票
	if command.Term == r.Term {
		if r.role == Candidate {
			resp.Term = r.Term
			resp.VoteGranted = false
			return
		}

		if r.role == Leader {
			resp.Term = r.Term
			resp.VoteGranted = false
			return
		}
	}

	// 请求投票节点任期比当前节点大

	// 正常逻辑一般都会到这里。
	// 领导人宕机后，第一个选举超时的节点，优先变为候选人。任期+1。然后发送rpc，其他节点还没超时，任期会小。
	if command.Term > r.Term {
		resp.Term = command.Term

		// 此时并不能返回给请求投票节点结果。因为日志还没验证，可能当前节点日志要比请求节点的日志更新。
		// 例如：A，B，C 三个节点。A节点为领导人，然后 A 节点宕机，B 节点先成为候选人并向 A，C 节点发送 RPC。
		// A节点宕机前，向 C 节点发送了最新的日志复制，并完成。此时 C 节点日志会比 B 的日志 更 新从而拒绝投票。
	}

	voteFor := r.storage.GetVoteFor() // 当前节点为其他节点投票

	if voteFor == command.CandidateId || voteFor == 0 {
		resp.VoteGranted = true
	}

	lastEntry, err := r.wal.lastEntry()
	if err != nil {
		resp.VoteGranted = false
		return
	}

	if lastEntry == nil {
		r.logger.Debug(">>> 拒绝投票 日志系统出错未能读取到最后一条日志", zap.Uint64("id", r.Id))
		err = errors.New(">>> 日志系统出错未能读取到最后一条日志")
		resp.VoteGranted = false
		return
	}

	if lastEntry.Index <= command.LastLogIndex && lastEntry.Term <= command.LastLogTerm {
		resp.VoteGranted = true
	} else {
		r.logger.Debug(">>> 拒绝投票 请求投票节点日志比较旧", zap.Uint64("id", r.Id), zap.Uint64("lastIndex", lastEntry.Index),
			zap.Uint64("commandLastIndex", command.LastLogIndex), zap.Uint64("lastTerm", lastEntry.Term),
			zap.Uint64("commandLastTerm", command.LastLogTerm))
	}

	if !resp.VoteGranted {
		return
	}

	r.logger.Debug(">>> 成功投票", zap.Uint64("candidateId", command.CandidateId), zap.Uint64("currentNode", r.Id))

	voteFor = command.CandidateId
	r.becomeFollower(resp.Term, 0, voteFor) // 当前节点变为跟随者

}

// receiveAppendEntriesRpc 收到追加日志请求
func (r *Raft) receiveAppendEntriesRpc(command *pb.AppendEntries, rpc *RPCReq) {
	r.logger.Debug(">>> 收到追加日志请求", zap.Uint64("leaderId", command.GetLeaderId()),
		zap.Uint64("leaderTerm", command.Term),
		zap.Uint64("currentTerm", r.Term))

	var err error
	resp := &pb.AppendEntriesResp{
		Term:    r.Term,
		Success: false,
	}

	defer func() {
		rpc.RespChan <- RPCResp{
			Err:  err,
			Resp: resp,
		}
	}()

	// Leader 任期小于
	// 说明 Leader 集群发生了网络分区。leader 节点与本节点断开了连接。
	// 本节点所在的集群已经重新选举出来了一个领导人
	if command.Term < r.Term {
		r.logger.Warn(">>> Leader 节点任期小于本节点", zap.Uint64("leaderId", command.GetLeaderId()))
		return
	}

	// Leader 任期大于当前任期
	if command.Term > r.Term {
		resp.Term = command.Term
	}

	// 当前节点非 Follower
	if r.role != Follower {
		r.becomeFollower(command.Term, command.LeaderId, 0)
	}

	// 重置选举定时
	r.electionCount = 0

	// 收到领导人发送过来的追加日志请求

	// 1.校验日志索引
	// 2.是否心跳
	// 3.校验日志冲突 删除冲突日志
	// 4.追加日志
	// 5.推进日志
	err = r.wal.appendEntries(command)
	if err == nil {
		resp.Success = true
	}
}

// electSelf 选举超时自身节点参与到选举中
func (r *Raft) electSelf() {
	r.electionCount++

	if r.electionCount >= r.randomElectionTimeout {
		r.logger.Debug(">>> 选举超时", zap.Int("electionCount", r.electionCount), zap.Int("randomElectionTimeout", r.randomElectionTimeout))

		// 选举超时重置计数器
		r.electionCount = 0

		if r.role == Leader {
			r.logger.Warn("current node role is leader ignore election", zap.Uint64("id", r.Id))
			return
		}

		r.becomeCandidate() // 成为候选人
	}
}

func (r *Raft) becomeFail() {
	r.role = Fail
}

func (r *Raft) becomeFollower(term uint64, leaderId uint64, voteFor uint64) {
	r.role = Follower
	r.Term = term
	r.LeaderId = leaderId
	r.electionCount = 0
	r.randomElectionTime() // 重置随机选举超时
	r.tick = r.electSelf

	for _, peer := range r.peers {
		peer.replicatingState = nil
	}

	if term != 0 {
		err := r.storage.SetTerm(term)
		if err != nil {
			r.becomeFail()
		}
	}
	if voteFor != 0 {
		err := r.storage.SetVotedFor(voteFor)
		if err != nil {
			r.becomeFail()
		}
	}
}

func (r *Raft) becomeCandidate() {
	r.role = Candidate
	r.Term++
	r.randomElectionTime() // 重置随机选举超时

	r.logger.Debug(">>> become candidate", zap.Uint64("id", r.Id), zap.Uint64("term", r.Term))

	// 最近的日志条目
	entry, err := r.wal.lastEntry()
	if err != nil {
		r.logger.Error("become candidate get last log entry fail", zap.Error(err))
		return
	}

	var lastLogIndex uint64
	var lastLogTerm uint64
	if entry != nil {
		lastLogIndex = entry.Index
		lastLogTerm = entry.Term
	}

	req := &pb.RequestVote{
		Term:         r.Term,
		CandidateId:  r.Id,
		LastLogIndex: lastLogIndex,
		LastLogTerm:  lastLogTerm,
	}

	r.grantedVotes = 0                                // 初始化投票计数
	r.voteRespCh = make(chan *voteResp, len(r.peers)) // 初始化投票结果 chan

	for id, e := range r.peers {
		if id == r.Id {
			// 为自己投票
			go func() {
				err := r.storage.SetTermAndVotedFor(r.Term, r.Id) // 持久化投票信息
				if err != nil {
					r.logger.Error("storage vote fail", zap.Error(err))
				}

				r.voteRespCh <- &voteResp{
					RequestVoteResp: pb.RequestVoteResp{
						Term:        r.Term,
						VoteGranted: true,
					},
					id: r.Id,
				}
			}()
		} else {
			// 请求其他节点为自己投票
			go func() {
				r.logger.Debug(">>> 发送请求投票请求", zap.Uint64("targetId", id), zap.String("targetAddress", e.endpoint.String()))

				voteResp := &voteResp{} // 投票结果

				err := r.trans.RequestVote(req, e.endpoint, &voteResp.RequestVoteResp)

				if err != nil {
					r.logger.Error("请求其他节点为自己投票失败 网络错误", zap.Uint64("id", r.Id), zap.String("endpoint", e.endpoint.String()), zap.Error(err))
					// 网络错误；投票肯定失败；
					voteResp.Term = r.Term
					voteResp.VoteGranted = false
					voteResp.id = e.id
				}

				r.voteRespCh <- voteResp
			}()
		}
	}
}

func (r *Raft) becomePreCandidate() {

}

func (r *Raft) becomeLeader() {
	r.logger.Debug(">>> become leader", zap.Uint64("id", r.Id))

	r.role = Leader
	r.LeaderId = r.Id

	r.tick = r.appendEntries // 定时发送追加日志请求；心跳

	// 初始化日志复制组件
	for _, peer := range r.peers {
		// lastIndex + 1
		// 日志复制组件为；leader 节点 下一条日志项索引；
		// 也就是说，leader 节点从 下一条日志项开始复制；
		peer.replicatingState = newReplicatingState(r.wal.lastIndex + 1)
	}

	// 写入空日志
	err := r.wal.appendEntry(&Entry{
		Kind:  KindNoOp,
		Index: r.wal.lastIndex + 1,
		Term:  r.Term,
	})
	if err != nil {
		r.logger.Error(">>> 领导人写入空日志出错，退化为跟随者", zap.Uint64("id", r.Id), zap.Error(err))
		r.becomeFollower(r.Term+1, 0, 0)
		return
	}

	r.appendEntries0() // 立即发送一次心跳
}

// appendEntries 追加日志
// Leader 定时追加日志
// 定时调度
func (r *Raft) appendEntries() {
	r.heartbeatCount++

	if r.heartbeatCount >= r.conf.HeartbeatTimeout/100 {
		r.heartbeatCount = 0

		if r.role != Leader {
			r.logger.Error("current node role is leader ignore send heartbeat", zap.Uint64("id", r.Id))
			return
		}

		r.appendEntries0()
	}
}

func (r *Raft) appendEntries0() {
	r.logRespCh = make(chan *logResp, len(r.peers))

	for _, peer := range r.peers {
		r.doAppendEntries(peer)
	}
}

// doAppendEntries 发送追加日志请求
func (r *Raft) doAppendEntries(peer *peer) {
	if r.Id == peer.id {
		lastEntry, err := r.wal.getEntry(r.wal.lastIndex)
		if err != nil {
			r.logger.Error(">>> 获取最后一条日志失败", zap.Uint64("id", r.Id), zap.Error(err))
			return
		}

		peer.replicatingState.matchIndex = lastEntry.Index // 领导节点直接复制成功
		peer.replicatingState.nextIndex = lastEntry.Index + 1
		peer.replicatingState.replicating = false
		return
	}

	go func() {
		r.logger.Debug(">>> 发送追加日志请求", zap.Uint64("id", r.Id))

		peer.replicatingState.replicating = true // 标记正在同步日志

		nextIndex := peer.replicatingState.nextIndex    // 准备传输的日志条目
		prevEntry, err := r.wal.getEntry(nextIndex - 1) // 获取传入日志条目的上一条
		if err != nil {
			r.logger.Error(">>> 获取上一条日志失败", zap.Uint64("index", nextIndex-1), zap.Error(err))
			return
		}

		// 组装需要传输的日志
		// XXX 优化点一次传输的太多或者太少都不好，没有判断日志的大小，全部传输了。
		entries := make([]*pb.Entry, 0)

		for i := nextIndex; i <= r.wal.lastIndex; i++ {
			e, err := r.wal.getEntry(i)
			if err != nil {
				r.logger.Error(">>> 获取日志失败", zap.Uint64("index", i), zap.Error(err))
			}

			entries = append(entries, &pb.Entry{
				Kind:    uint64(e.Kind),
				Index:   e.Index,
				Term:    e.Term,
				Command: e.Command,
			})
		}

		// 根据replicatingState日志复制状态，来确定发送的日志条目
		req := &pb.AppendEntries{
			Term:         r.Term,
			LeaderId:     r.Id,
			PrevLogIndex: prevEntry.Index,
			PrevLogTerm:  prevEntry.Term,
			LeaderCommit: r.wal.commitIndex,
			Entries:      entries,
		}

		var resp = &logResp{
			id: peer.id, // leader 需要知道是哪个节点回复的消息
		}

		if err := r.trans.AppendEntries(req, peer.endpoint, &resp.AppendEntriesResp); err != nil {
			// 网络错误
			r.logger.Error("发送追加日志失败 网络错误", zap.Uint64("leaderId", r.Id), zap.String("endpoint", peer.endpoint.String()), zap.Error(err))
			resp.Term = r.Term
			resp.Success = false
		}

		r.logRespCh <- resp
	}()
}

// handleAppendEntriesResp 收到日志回复结果
func (r *Raft) handleAppendEntriesResp(resp *logResp) {
	r.logger.Debug(">>> 收到日志回复结果", zap.Uint64("fromId", resp.id))

	// leader 任期小于其他节点
	if resp.Term > r.Term && !resp.Success {
		r.becomeFollower(resp.Term, 0, 0)
		return
	}

	if r.role != Leader {
		// 发生网络分区；leader 与 一部分 follower 分了一个区，另外的 follower 在另外的区 并 重新选举出了一个 leader；
		// 网络分区修复后。old leader 发送心跳给其他节点，new follower 和 new leader 肯定会回复更高的任期。
		// 第一个心跳回复之后，old leader 就退化为 follower 了。
		// 后续的请求，就打印点日志的了。
		r.logger.Warn("非领导人节点收到了心跳回复", zap.Uint64("id", r.Id), zap.Uint64("FromId", resp.id))
		return
	}

	// 当前节点为 Leader 节点。收到 Follower 回复。日志是否追加成功。来判断是否多数成功，来推进本地日志。还有就是如果不成功，怎么弥补冲突日志。

	peer := r.peers[resp.id] // 回复的节点
	leaderPeer := r.peers[r.Id]
	if resp.Success {
		// 回复成功说明复制成功了。
		peer.replicatingState.matchIndex = leaderPeer.replicatingState.matchIndex
		peer.replicatingState.nextIndex = leaderPeer.replicatingState.nextIndex
	} else {
		peer.replicatingState.nextIndex-- // 复制索引往回退一步
	}

	// 获取推进日志索引
	indexes := make([]uint64, 0)
	for _, p := range r.peers {
		if p.id != r.Id {
			entry, err := r.wal.getEntry(p.replicatingState.matchIndex)
			if err != nil {
				r.logger.Error(">>> getEntry is err", zap.Uint64("id", p.id), zap.Error(err))
				return
			}

			if entry.Term == r.Term { // 只收集任期相同
				indexes = append(indexes, p.replicatingState.matchIndex)
			}
		}
	}

	// 最正常的情况，直接复制 4,5,6
	// B C 节点 matchIndex 都为 6。 indexes=[6,6] 取中位数，为6，推进日志为 6
	// A leader 	[1,1][1,2][1,3][1,4][1,5][1,6]
	// B follower	[1,1][1,2][1,3]
	// C follower	[1,1][1,2][1,3]

	// B 节点复制成功。 matchIndex=6
	// C 节点复制失败。 matchIndex=0
	// indexes=[6,0] 取中位数为 0 ，推进日志为 0
	// 这种情况是不合理的。RAFT 算法规定只能提交相同任期内的日志。所以，C 节点复制失败 matchIndex=0,任期不同。
	// indexes=[6] 推进 6.
	// A leader 	[1,1][1,2][1,3][2,4][2,5][2,6]
	// B follower	[1,1][1,2][1,3]
	// C follower	[1,1][1,2][2,3]

	// 降序
	sort.Slice(indexes, func(i, j int) bool { return indexes[i] > indexes[j] })

	commitIndex := indexes[len(indexes)/2] // 取中位数

	err := r.wal.advanceCommitIndex(commitIndex, r.Term)
	if err != nil {
		r.logger.Error(">>> commit log is err", zap.Uint64("id", r.Id), zap.Error(err))
	}
}

// randomElectionTime 随机选举时间
func (r *Raft) randomElectionTime() {
	// 100 为 100 毫秒的 tick
	// 3000 毫秒的超时时间; tick 每 100 毫秒，触发一次，3000/100 触发 30 次。正好超时。
	r.randomElectionTimeout = (rand.IntN(r.conf.MaxElectionTimeout-r.conf.MinElectionTimeout) + r.conf.MinElectionTimeout) / 100
}
